package com.wechat.network.peer;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean;

import com.swak.exception.InvokeException;
import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.ClusterConfig;
import com.swak.reactivex.cluster.ClusterExchangeHandler;
import com.swak.reactivex.cluster.ID;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.future.Futures;
import com.swak.reactivex.transport.Message;
import com.swak.utils.Ints;
import com.swak.utils.Lists;
import com.swak.utils.StringUtils;
import com.wechat.Cons;
import com.wechat.command.Command;
import com.wechat.command.CommandResult;
import com.wechat.command.Commands;
import com.wechat.login.Login;
import com.wechat.login.LoginService;
import com.wechat.network.AbstractNetwork;
import com.wechat.network.Network;
import com.wechat.network.NetworkHandler;
import com.wechat.network.NetworkHandler.NetworkExchangeHandler;
import com.wechat.ser.result.FindResult;
import com.wechat.ser.result.GetInfoResult;

/**
 * P2P 网络
 * 
 * @author 618lf
 */
public class PeerNetwork extends AbstractNetwork implements Network {

	private static final String Riad = "BeeChat";

	private final AtomicBoolean closed = new AtomicBoolean(false);
	private final NetworkHandler handler;

	private final Login login;
	private Member member;
	private ClusterConfig config;

	public PeerNetwork(String ser, NetworkHandler handler) {
		this.handler = handler;
		this.login = LoginService.login();
		this.config = new ClusterConfig();
		if (StringUtils.isNotBlank(ser)) {
			this.config.getDiscovery().setMulticastPort(Integer.parseInt(ser));
		}
		this.config.setName(Riad);
		this.config.setPeerID(login.getId());
	}

	@Override
	public void connect() throws RemotingException {
		if (this.closed.get()) {
			return;
		}
		try {
			this.member = Cluster.join(this.config, new ChatExchangeHandler(handler));
		} catch (Exception e) {
			this.closed.set(true);
			throw new RemotingException("connect to p2p network error.", e);
		}
	}

	@Override
	public CompletionStage<CommandResult> send(Command command) {

		if (logger.isDebugEnabled()) {
			logger.debug("[网络] 发送命令 {}", command.getData());
		}

		// 网络已经关闭
		if (this.closed.get()) {
			return CompletableFuture.completedFuture(CommandResult.fail(StringUtils.format("网络已经关闭！")));
		}

		// 调用服务(拦截了服务)
		if (command.getCommand() == Commands.Ser_Find || command.getCommand() == Commands.Ser_Login) {
			return this.invokeSer(command);
		}

		// 发送消息
		return this.sendMessage(command);
	}

	// 调用服务
	private CompletionStage<CommandResult> invokeSer(Command command) {
		if (command.getCommand() == Commands.Ser_Login) {
			return this.invokeSer_Login(command);
		} else if (command.getCommand() == Commands.Ser_Find) {
			return this.invokeSer_Find(command);
		}
		return CompletableFuture.completedStage(CommandResult.fail("Command not support!"));
	}

	private CompletionStage<CommandResult> invokeSer_Login(Command command) {
		return CompletableFuture.completedStage(CommandResult.success());
	}

	private CompletionStage<CommandResult> invokeSer_Find(Command command) {
		List<CompletableFuture<CommandResult>> futures = Lists.newArrayList();
		List<Member> members = this.member.members().members();
		for (Member member : members) {
			if (member.isThis()) {
				continue;
			}
			futures.add(this.doSendMessageToPeer(member,
					Command.cmdForGetInfo(this.member.id().getId(), member.id().getId())));
		}
		return Futures.allResult(futures).thenApply(results -> {
			List<FindResult> finds = Lists.newArrayList();
			for (CommandResult result : results) {
				if (result.isSuccess()) {
					GetInfoResult info = result.result2Object();
					finds.add(new FindResult().setId(info.getId()).setName(info.getName()));
				}
			}
			return finds;
		}).thenApply(res -> CommandResult.success(res));
	}

	// 发布消息
	private CompletionStage<CommandResult> sendMessage(Command message) {
		if (StringUtils.endsWith(message.getTo(), Cons.ID_ChatRoom)) {
			return this.sendMessageToRoom(message);
		}
		return this.sendMessageToPeer(message);
	}

	// 发布消息 - 发送给群组
	private CompletionStage<CommandResult> sendMessageToRoom(Command message) {
		return this.nodeService.getChatRoom(message.getTo()).thenApply(room -> {

			// 顺序的发送数据
			try {
				String[] userNames = room.getUserNames().split("\\" + Cons.UserNamesSeparateChar);
				for (String userName : userNames) {
					this.sendMessageToPeer(message.copy().setTo(userName));
				}
			} catch (Exception e) {
				// ig
			}

			// 发送成功
			return CommandResult.success();
		});
	}

	// 发布消息 - 发送给个人
	private CompletionStage<CommandResult> sendMessageToPeer(Command message) {

		// 节点
		String peerID = message.getTo();

		// 检查节点
		if (StringUtils.endsWith(message.getTo(), Cons.ID_ChatRoom) // room
				|| StringUtils.endsWith(message.getTo(), Cons.ID_Robot) // robot
				|| login.getId().equals(peerID)) {// me
			return CompletableFuture.completedFuture(CommandResult.fail(StringUtils.format("不能发送消息给节点[%s]！", peerID)));
		}

		// 查找节点
		Member member = this.member.members().lookup(new ID().setId(peerID));
		if (member == null) {
			return CompletableFuture.completedFuture(CommandResult.fail(StringUtils.format("节点[%s] 已经下线！", peerID)));
		}

		// 通过 TCP 发送数据
		return this.doSendMessageToPeer(member, message);
	}

	private CompletableFuture<CommandResult> doSendMessageToPeer(Member member, Command message) {
		CompletableFuture<CommandResult> resultFuture = new CompletableFuture<>();
		this.retryExecuteRunnable(() -> {
			try {
				CompletableFuture<Object> future = member.request(message);
				future.whenComplete((r, e) -> {
					if (e != null) {
						resultFuture.completeExceptionally(e);
					} else if (r != null && r instanceof Message) {
						resultFuture.complete(CommandResult.wrap(((Message) r).getData()));
					} else {
						resultFuture.complete(CommandResult.wrap(r));
					}
				});
			} catch (Throwable e) {
				throw new InvokeException("Send message error", e);
			}
		});
		return resultFuture;
	}

	private void retryExecuteRunnable(Runnable runnable) {
		this.retryExecuteRunnable(3, runnable);
	}

	private void retryExecuteRunnable(int times, Runnable runnable) {
		int tries = Ints.max(times, 3);
		while (tries > 0) {
			try {
				runnable.run();
				break;
			} catch (Exception e) {
			}
			tries--;
		}
	}

	@Override
	public void close() {
		if (this.closed.get()) {
			return;
		}
		this.closed.set(true);
		try {
			this.member.close();
		} catch (Exception e) {
		}
	}

	/**
	 * Chat 命令处理
	 * 
	 * @author 618lf
	 */
	class ChatExchangeHandler extends NetworkExchangeHandler implements ClusterExchangeHandler {

		ChatExchangeHandler(NetworkHandler delegate) {
			super(PeerNetwork.this, delegate);
		}

		@Override
		public void memberShipChanaged(Member change, List<Member> members, boolean add) {
		}
	}
}